package com.ksc.wordcount.datasourceapi;

import java.io.IOException;
import java.io.Serializable;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.stream.Stream;

/**
 * TextPartionReader是一个实现了PartionReader接口的类，用于读取文本文件的内容。
 * 它可以将一个或多个文件分片（FileSplit）中的所有行（line）合并成一个流（Stream）返回。
 */
public class TextPartionReader implements PartionReader<String>, Serializable {

    /**
     * 将PartionFile中的文件分片逐个读取，并将所有行合并成一个流（Stream）返回。
     * @param partionFile 包含文件分片的PartionFile对象
     * @return 一个包含所有行的流（Stream）
     * @throws IOException 如果读取文件时发生I/O错误
     */
    @Override
    public Stream<String> toStream(PartionFile partionFile) throws IOException {
        // 创建一个空流
        Stream<String> allStream = Stream.empty();

        // 遍历每个文件分片
        // todo 实现 maptask读取原始数据文件的内容
        for (FileSplit fileSplit : partionFile.getFileSplits()) {
            // 从文件中逐行读取，并将每行添加到allStream中
            long start = fileSplit.getStart();
            long length = fileSplit.getLength();

            // 只读取start偏移量到start+length的内容到linesStream中
            Stream<String> linesStream = Files.lines(Paths.get(fileSplit.getFileName()))
                    .skip(start)
                    .limit(length);

            allStream = Stream.concat(allStream, linesStream); // 将当前文件分片的行合并到allStream中
        }

        // 返回包含所有行的流
        return allStream;
    }
}